package com.ek.service.mina.handler;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ek.common.MessageHandler;
import com.ek.kademlia.protocol.Message;
import com.ek.kademlia.protocol.Response;

/**
 * @author G.Santoro
 */
public class DispatcherHandler extends IoHandlerAdapter {
    private static final transient Logger log = LoggerFactory.getLogger(DispatcherHandler.class);
    private Map<String, MessageHandler> handlers;
    private final Map<String, BlockingQueue<Response>> listeners = new ConcurrentHashMap<String, BlockingQueue<Response>>();

    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        log.debug("messageReceived.");

        if (message instanceof Response) {
            log.debug("the message is a response.");
            Response response = (Response) message;
            response.setSocketAddress(session.getRemoteAddress());
            BlockingQueue<Response> listener = listeners.get(new String(((Response) message).getMessageID()));
            if (listener != null) {
                listener.put(response);
            }

        } else if (message instanceof Message) {
            log.debug("the message is a message.");
            synchronized (handlers) {
                MessageHandler messageHandler = handlers.get(((Message) message).getMessageClass());
                if (messageHandler != null) {
                    messageHandler.execute(message, session.getRemoteAddress());
                } else {
                    throw new IllegalArgumentException("The message received:" + message + " don't have a relative Handler!");
                }
            }

        } else {
            throw new IllegalArgumentException("The message received:" + message + " don't implements Message interface!");

        }
    }

    public void setHandlers(Map<String, MessageHandler> handlers) {
        this.handlers = handlers;
    }

    public void putListener(String key, BlockingQueue<Response> listener) {
        listeners.put(key, listener);
    }

    public void removeListener(String key) {
        listeners.remove(key);
    }
}
